package com.mytx.search.redisMessageQueue;

import com.alibaba.fastjson.JSONObject;
import com.mytx.common.constant.Constant;
import com.mytx.pojo.LatLng;
import com.mytx.pojo.Topic;
import com.mytx.search.config.EsSearchClient;
import com.taobao.api.TaobaoClient;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.GeoDistanceQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.*;


@SuppressWarnings("Duplicates")
@Component
public class Receiver implements MessageListener {
    private static final Logger logger = LoggerFactory.getLogger(Receiver.class);

    //    @Value("${search_distance}") 不知道为什么在这儿 获取这个值无法启动
//    int searchDistance;
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    private TaobaoClient taobaoClient;
    @Autowired
    EsSearchClient searchClient;

    @Override
    public void onMessage(Message message, byte[] pattern) {

        String deserialize = new String(message.getBody(), Charset.forName("UTF-8"));
        String channel = new String(message.getChannel(), Charset.forName("UTF-8"));
        logger.info("onMessage:channel{},deserialize:{}", channel, deserialize);

        JSONObject info = JSONObject.parseObject(deserialize);
        switch (channel) {
            case Constant.NOTIFY_NEW_TOPIC: {//根据主题的地理位置，搜索需要推送的对象
                com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
                notify.setCreated(new Date());
                notify.setNick("系统消息");
                Topic topic = info.getObject("topic", Topic.class);
                LatLng location = topic.getLocation();
                List<String> userIdList = nearbySearch(location);
                if (userIdList.size() != 0) {
                    // userIdList.remove(topic.getManager());//从推送对象id中清除掉主题发布者，不然自己发的推送给自己，不科学
                    notify.setExtraData(topic);
                    notify.setType(Constant.PUSH_MESSAGE_NEW_TOPIC);
                    pushMessage(userIdList, notify);
                }


                break;
            }
            case Constant.NOTIFY_FOUND_COUPON: {
                com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
                notify.setCreated(new Date());
                notify.setNick("系统消息");
                String topicId = info.getString("topicId");
                if (topicId == null) {
                    return;
                }
                Topic topic = getTopicById(topicId);
                LatLng location = topic.getLocation();
                List<String> userIdList = nearbySearch(location);
                if (userIdList.size() != 0) {
                    // userIdList.remove(topic.getManager());//从推送对象id中清除掉主题发布者，不然自己发的推送给自己，不科学
                    notify.setExtraData(topic);
                    notify.setType(Constant.PUSH_MESSAGE_FOUND_COUPON);
                    pushMessage(userIdList, notify);
                }
                break;
            }

            case Constant.NOTIFY_OFFER_TO_TOPIC_FAIL: {
                com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
                notify.setCreated(new Date());
                notify.setNick("系统消息");
                List userIdList = info.getJSONArray("userIdList").toJavaList(String.class);
                notify.setExtraData(info.get("topic"));
                notify.setType(Constant.PUSH_MESSAGE_OFFER_TO_TOPIC_FAIL);
                pushMessage(userIdList, notify);
                break;
            }
            case Constant.NOTIFY_ORDER_INFO:
                pushOrderInfo(info, Constant.PUSH_MESSAGE_ORDER_INFO);
                break;
            case Constant.NOTIFY_TASK_INFO:
                pushTaskInfo(info, Constant.PUSH_MESSAGE_TASK_INFO);
                break;

            case Constant.NOTIFY_TOPIC_VERIFY:
                com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
                notify.setCreated(new Date());
                notify.setNick("系统消息");
                List<String> userIdList = new ArrayList();
                userIdList.add(info.getString("uId"));
                notify.setExtraData(info.get("topic"));
                notify.setVisible(Constant.MESSAGE_VISIBLE);
                notify.setType(Constant.PUSH_MESSAGE_TOPIC_VERIFY);
                notify.setCreated(new Date());
                pushMessage(userIdList, notify);
                break;


        }
    }

    private void pushOrderInfo(JSONObject info, int pushMessageOrderRefuseRefund) {
        com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
        notify.setCreated(new Date());
        notify.setNick("系统消息");
        notify.setExtraData(info.get("order"));
        List<String> userIdList = new ArrayList<>();
        userIdList.add((String) info.remove("uId"));
        notify.setType(pushMessageOrderRefuseRefund);
        pushMessage(userIdList, notify);
    }

    private void pushTaskInfo(JSONObject info, int pushMessageOrderRefuseRefund) {
        com.mytx.pojo.Message notify = new com.mytx.pojo.Message();
        notify.setCreated(new Date());
        notify.setNick("系统消息");
        notify.setExtraData(info.get("task"));
        List<String> userIdList = new ArrayList<>();
        if (info.containsKey("uId")) {
            userIdList.add((String) info.remove("uId"));
        } else if (info.containsKey("userIdList")) {
            userIdList.addAll(info.getJSONArray("userIdList").toJavaList(String.class));
        }
        notify.setType(pushMessageOrderRefuseRefund);
        pushMessage(userIdList, notify);
    }

    public List<String> nearbySearch(LatLng location) {
        List<String> userIdList = new ArrayList<>();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        TermQueryBuilder statusQueryBuilder = QueryBuilders.termQuery("type", Constant.TOPIC_TYPE_USER_INFO);
        boolQueryBuilder.filter(statusQueryBuilder);
        GeoDistanceQueryBuilder distanceQueryBuilder = QueryBuilders.geoDistanceQuery("location")
                .distance(5, DistanceUnit.KILOMETERS)
                .point(location.getLat(), location.getLon());
        boolQueryBuilder.must(distanceQueryBuilder);
        searchSourceBuilder.query(boolQueryBuilder);
        SearchRequest searchRequest = new SearchRequest("topics");//topics是别名,索引名是 topic_v1,类型是：topic
        searchRequest.types("topic");
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = searchClient.search(searchRequest);

        SearchHits hits = searchResponse.getHits();

        for (SearchHit searchHit : hits.getHits()) {
            userIdList.add(searchHit.getId());
        }

        return userIdList;
    }

    private void pushMessage(List<String> userIdList, com.mytx.pojo.Message notify) {
        JSONObject info = new JSONObject();
        info.put("userIdList",userIdList);
        info.put("message",notify);
        redisTemplate.convertAndSend(Constant.NOTIFY_PUSH_MESSAGE,info );
    }

    private Topic getTopicById(String topicId) {

        GetRequest request = new GetRequest("topics", "topic", topicId);
        GetResponse response = searchClient.get(request);
        if (response == null) {
            return null;
        }
        return JSONObject.parseObject(JSONObject.toJSONString(response.getSource()), Topic.class);

    }
}
